package org.zjt.weather.demo;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.io.DoubleWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

/**
 * DESC 接受myGroup之后的数据，返回数据存放到hdfs中。
 *
 * @author
 * @create 2017-05-23 下午2:32
 **/
public class WeatherReducer extends Reducer<WeatherKey, DoubleWritable, Text, Text> {

    /**
     * @param key     得到相同年月的key
     * @param values  相同年月的value数组
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    protected void reduce(WeatherKey key, Iterable<DoubleWritable> values, Context context) throws IOException, InterruptedException {

        List<Double> list = new ArrayList<>();
        values.forEach(a -> list.add(a.get()));
        List<String> collect = list.stream().sorted().limit(3).map(a -> a.toString() + " 摄氏度").collect(Collectors.toList());
        String text = StringUtils.join(collect, ",");
        String key1 = String.format("%s-%s :", key.getYear(), key.getMonth());
        context.write(new Text(key1), new Text(text));
        //list.sort((a,b)-> a > b ? 1 : a = b ? 0 : -1 );
    }
}
